package com.gis.bigdata.spark02.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author LnnuUser
 * @create 2022-10-06-下午12:02
 */
object Scala_Test03_1 {
  // 需求3：页面单跳转换率统计

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("flowApp")
    val sc = new SparkContext(conf)

    val actionRDD: RDD[String] = sc.textFile("/home/lnnu/IdeaProjects/spark/datas/user_visit_action.txt")

    val actionDataRDD: RDD[UserVisitAction] = actionRDD.map(
      action => {
        val datas: Array[String] = action.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )

    actionDataRDD.cache()

    // 对指定的页面连续跳转进行统计
    val ids = List(1,2,3,4,5,6,7)

    // 1. 计算分母
    val pageidToCountMap: Map[Long, Long] = actionDataRDD.map(
      action => {
        (action.page_id, 1L)
      }
    ).reduceByKey(_ + _).collect().toMap

    // 2. 计算分子
    // 根据session进行分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)

    // 分组后，根据访问时间进行排序（升序）
    val mvRDD: RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
      iter => {
        val sortList: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
        val flowIds: List[Long] = sortList.map(_.page_id)
        // [1,2,3,4]
        // [2,3,4]
        // zip -> [1-2,2-3,3-4]
        val pageflowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)
        pageflowIds.map(
          t => {
            (t, 1)
          }
        )
      }
    )

    // ((1,2), 1)
    val flatRDD: RDD[((Long, Long), Int)] = mvRDD.map(_._2).flatMap(list => list)

    // ((1,2), sum)
    val dataRDD: RDD[((Long, Long), Int)] = flatRDD.reduceByKey(_ + _)

    // 3. 计算单挑转换率
    // 分子除以分母
    dataRDD.foreach{
      case ((pageid1, pageid2), sum ) => {
        val lon: Long = pageidToCountMap.getOrElse(pageid1, 0L)
        println(s"页面${pageid1}跳转到页面${pageid2}单挑转换率为:" + (sum.toDouble/lon))
      }
    }

    sc.stop()

  }

  case class UserVisitAction(
                            date:String,
                            user_id: Long,
                            session_id: String,
                            page_id: Long,
                            action_time: String,
                            search_keyword: String,
                            click_category_id: Long,
                            click_product_id: Long,
                            order_category_ids: String,
                            order_product_ids: String,
                            pay_categroy_ids: String,
                            pay_product_ids: String,
                            city_id: Long
                            )

}
